
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import exception.IllegalObjectIdException;
import indi.hjhk.csv.CSVReader;
import indi.hjhk.exception.ExceptionSerializer;
import indi.hjhk.log.Logger;
import nju.hjh.arcadedb.timeseries.DataType;
import nju.hjh.arcadedb.timeseries.TimeseriesEngine;
import nju.hjh.arcadedb.timeseries.datapoint.StringDataPoint;
import nju.hjh.arcadedb.timeseries.exception.DuplicateTimestampException;
import nju.hjh.arcadedb.timeseries.exception.TimeseriesException;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

public class CarTsTransServerTest {
    public static void main(String[] args) {
        Logger logger = Logger.getPureLogger("carTrans");

        ArcadeTSDBClient client = new ArcadeTSDBClient();

        long totalSuccess = 0, totalCsvElapsed = 0, totalJsonBuildElapsed = 0, totalServerResponseElapsed = 0;

        ArrayList<String> csvFiles = new ArrayList<>();
        csvFiles.add("/hdd/data/1101/1101_sort.csv");

        JSONObject jsonInsert = new JSONObject();
        jsonInsert.put("database", "carTest");
        jsonInsert.put("action", "insert");

        HashMap<String, JSONArray> timeseriesList = new HashMap<>();

        final int jsonSize = 10000, statSize = 300000;
        logger.logOnStdout("json size= %d, stat size= %d", jsonSize, statSize);

        for (String csvFilePath : csvFiles) {
            logger.appendTag("csvFile", () -> String.format("csv: %s", csvFilePath));
            long offset = 0;

            CSVReader csvReader;
            try {
                csvReader = new CSVReader(csvFilePath);
                csvReader.readHeader();
            }catch (IOException e){
                logger.logOnStderr("cannot read csv file %s, skipped", csvFilePath);
                logger.logException(e);
                continue;
            }catch (CSVReader.CSVFormatException e){
                logger.logOnStderr("csv file %s has no header line, skipped", csvFilePath);
                logger.logException(e);
                continue;
            }

            long startTime, csvElapsed = 0, jsonBuildElapsed = 0, serverResponseElapsed = 0;
            long subCsvElapsed = 0, subJsonBuildElapsed = 0, subServerResponseElapsed = 0;

            logger.logOnStdout("handling data within file %s:", csvFilePath);

            int successCount = 0, buildCount = 0;

            try {
                while (csvReader.next()) {
                    startTime = System.currentTimeMillis();
                    String carId = csvReader.getString("VLP") + "_" + csvReader.getString("VLPC");
                    long timestamp = csvReader.getLong("TIME");
                    String passId = csvReader.getString("PASSID");
                    String stationId = csvReader.getString("STATIONID");

                    subCsvElapsed += System.currentTimeMillis() - startTime;
                    startTime = System.currentTimeMillis();

                    JSONArray timeseries = timeseriesList.get(carId);
                    if (timeseries == null){
                        timeseries = new JSONArray();
                        timeseriesList.put(carId, timeseries);
                    }

                    boolean exist = false;
                    for (Object obj : timeseries){
                        if (obj instanceof JSONObject metric){
                            if (metric.getLong("time") == timestamp){
                                exist = true;
                                break;
                            }
                        }
                    }

                    if (!exist){
                        JSONObject newMetric = new JSONObject();
                        newMetric.put("time", timestamp);

                        JSONObject jsonMetrics = new JSONObject();
                        jsonMetrics.put("stationId", stationId);
                        jsonMetrics.put("passId", passId);

                        newMetric.put("metrics", jsonMetrics);
                        timeseries.add(newMetric);
                    }

                    subJsonBuildElapsed += System.currentTimeMillis() - startTime;
                    buildCount++;

                    if (buildCount >= jsonSize) {
                        startTime = System.currentTimeMillis();
                        JSONArray inserts = new JSONArray();
                        for (Map.Entry<String, JSONArray> entry : timeseriesList.entrySet()){
                            String car = entry.getKey();
                            JSONObject insert = new JSONObject();
                            insert.put("objectid", car);
                            insert.put("timeseries", entry.getValue());
                            inserts.add(insert);
                        }
                        jsonInsert.put("inserts", inserts);
                        subJsonBuildElapsed += System.currentTimeMillis() - startTime;
                        timeseriesList.clear();

                        startTime = System.currentTimeMillis();
                        String ret = client.sendJsonAndWaitResult(jsonInsert);
                        JSONObject retObject = JSONObject.parseObject(ret);

                        boolean success = true;
                        if (retObject.containsKey("exceptions")) {
                            success = false;
                            logger.logOnStderr(ArcadeTSDBClient.toPrettyPrintJSON(retObject.getJSONArray("exceptions").toJSONString()));
                        }
                        else{
                            for (Map.Entry<String, Object> entry: retObject.entrySet()){
                                if (entry.getValue() instanceof JSONObject result)
                                    if (!result.getBoolean("success")){
                                        JSONObject exception = result.getJSONObject("exception");
                                        if (exception == null)
                                            continue;

                                        String exceptionClass = exception.getString("class");
                                        try {
                                            if (Class.forName(exceptionClass).equals(IllegalObjectIdException.class)) {
                                                logger.logOnStderr("invalid car id %s found, skipped", entry.getKey());
                                                continue;
                                            }
                                        }catch (ClassNotFoundException ignored){}

                                        success = false;
                                        logger.logOnStderr(ArcadeTSDBClient.toPrettyPrintJSON(exception.toJSONString()));
                                        break;
                                    }
                            }
                        }
                        if (!success){
                            throw new Exception("server return failure");
                        }
                        subServerResponseElapsed += System.currentTimeMillis() - startTime;

                        successCount += buildCount;
                        buildCount = 0;
                    }

                    if (successCount >= statSize){
                        csvElapsed += subCsvElapsed;
                        jsonBuildElapsed += subJsonBuildElapsed;
                        serverResponseElapsed += subServerResponseElapsed;
                        offset += successCount;

                        logger.logOnStdout("successfully committed %d data (end index= %d), timeSpent: CSV: %dms + JSON: %dms + ArcadeTSDBServer: %dms = %dms", successCount, offset - 1, subCsvElapsed, subJsonBuildElapsed, subServerResponseElapsed, subCsvElapsed + subServerResponseElapsed + subJsonBuildElapsed);

                        successCount = 0;
                        subCsvElapsed = 0;
                        subServerResponseElapsed = 0;
                        subJsonBuildElapsed = 0;
                    }
                }

                startTime = System.currentTimeMillis();
                JSONArray inserts = new JSONArray();
                for (Map.Entry<String, JSONArray> entry : timeseriesList.entrySet()){
                    String car = entry.getKey();
                    JSONObject insert = new JSONObject();
                    insert.put("objectid", car);
                    insert.put("timeseries", entry.getValue());
                    inserts.add(insert);
                }
                jsonInsert.put("inserts", inserts);
                subJsonBuildElapsed += System.currentTimeMillis() - startTime;
                timeseriesList.clear();

                startTime = System.currentTimeMillis();
                String ret = client.sendJsonAndWaitResult(jsonInsert);
                JSONObject retObject = JSONObject.parseObject(ret);

                boolean success = true;
                if (retObject.containsKey("exceptions")) {
                    success = false;
                    logger.logOnStderr(ArcadeTSDBClient.toPrettyPrintJSON(retObject.getJSONArray("exceptions").toJSONString()));
                }
                else{
                    for (Map.Entry<String, Object> entry: retObject.entrySet()){
                        if (entry.getValue() instanceof JSONObject result)
                            if (!result.getBoolean("success")){
                                success = false;
                                logger.logOnStderr(ArcadeTSDBClient.toPrettyPrintJSON(result.getJSONObject("exception").toJSONString()));
                                break;
                            }
                    }
                }
                if (!success){
                    throw new Exception("server return failure");
                }
                subServerResponseElapsed += System.currentTimeMillis() - startTime;

                csvElapsed += subCsvElapsed;
                jsonBuildElapsed += subJsonBuildElapsed;
                serverResponseElapsed += subServerResponseElapsed;
                successCount += buildCount;
                offset += successCount;

                logger.logOnStdout("successfully committed %d data (end index= %d), timeSpent: CSV: %dms + JSON: %dms + ArcadeTSDBServer: %dms = %dms", successCount, offset - 1, subCsvElapsed, subJsonBuildElapsed, subServerResponseElapsed, subCsvElapsed + subServerResponseElapsed + subJsonBuildElapsed);

                totalCsvElapsed += csvElapsed;
                totalJsonBuildElapsed += jsonBuildElapsed;
                totalServerResponseElapsed += serverResponseElapsed;
                totalSuccess += offset;

            } catch (Exception e) {
                client.close();
                logger.logOnStderr("Problem occurred when transferring to ArcadeDB at offset %d (+%d)", offset, successCount);
                logger.logException(e);
                return;
            }

            logger.logOnStdout(String.format("Transferred to ArcadeDB(success: %d), CSV: %dms + JSON: %dms + ArcadeTSDBServer: %dms = %dms", offset, csvElapsed, jsonBuildElapsed, serverResponseElapsed, csvElapsed + jsonBuildElapsed + serverResponseElapsed));
            logger.removeTag("csvFile");
        }
        logger.logOnStdout(String.format("Total: Transferred from %d csv file(success: %d) to ArcadeDB, CSV: %dms + JSON: %dms + ArcadeTSDBServer: %dms = %dms", csvFiles.size(), totalSuccess, totalCsvElapsed, totalJsonBuildElapsed, totalServerResponseElapsed, totalCsvElapsed + totalJsonBuildElapsed + totalServerResponseElapsed));
        client.close();
    }
}
